set spark.sql.crossJoin.enabled=true;
with cust_way as (
    select
        cust.date_time
         ,cust.customer_code
         ,cust.customer_name
         ,cust.customer_create_time
         ,cust.staff_code
         ,cust.staff_name
         ,belong_type
         ,cust.network_code
         ,cust.network_name
         ,nvl(way.waybill_cnt,0) as waybill_cnt
         ,cust.ordersource_name as ordersource_name --订单来源
    from
        ( --客户、市场业务员、所属类别关系
            select
                dt.date_mid_desc as date_time
                 ,cu.code  as customer_code
                 ,cu.name as customer_name
                 ,cu.create_time as customer_create_time
                 ,nvl(cu.market_staff_code,'') as staff_code
                 ,staff2.name as staff_name
                 ,staff.belong_type as belong_type
                 ,net.code as network_code
                 ,net.name as network_name
                 ,if(cu.source_id in ('326','348','352','353','328'),source_id,'其他') AS ordersource_name
            from jms_dim.dim_lmdm_sys_customer cu
                     inner join jms_dim.dim_date_info_base dt on dt.date_mid_desc between add_months(trunc('{{ execution_date | cst_ds}}','MM'),-3) and '{{ execution_date | cst_ds}}'  and 1=1
                     left join jms_dim.dim_yl_nwm_cost_config staff on staff.config_type_code=3 and dt.date_mid_desc between staff.start_time and staff.end_time and staff.code=cu.market_staff_code
                     left join jms_dim.dim_lmdm_sys_staff staff2 on staff2.code=cu.market_staff_code
                     left join jms_dim.dim_network_whole_massage net on net.id=cu.network_id
            where cu.is_enable=1 and cu.is_delete=1 and date(cu.create_time)<='{{ execution_date | cst_ds}}'
    )cust
    left join  ( --客户件量统计
select
    count(waybill_no) waybill_cnt,
    date(input_time) input_time,
    customer_code
from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
where
    dt between add_months(trunc('{{ execution_date | cst_ds}}','MM'),-3) and '{{ execution_date | cst_ds}}'
  and nvl(customer_code, '-') <> '-'
  and nvl(is_void, 0) = 0
  and is_delete <> 2
group by date(input_time),customer_code
    )way on cust.date_time=way.input_time and cust.customer_code=way.customer_code
    )

insert overwrite table jms_dm.dm_network_customer_analysis_detail_dt partition(dt)
select
 t3.date_time     --查询日期
,t3.customer_code --客户编码
,t3.customer_name --客户名称
,t3.customer_create_time --客户创建时间
,t3.staff_code     --员工编码
,t3.staff_name     --员工名称
,t3.belong_type    --所属类别
,t3.network_code   --网点编码
,t3.network_name   --网点名称
,t3.waybill_cnt     --当日发货量
,cast(t3.mm_waybill_cnt/send_days  as decimal(16,0)) as avg_waybill_cnt  --日均发货
,case when cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))>5000 then 'A'
      when cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))<=5000 and cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))>3000 then 'B+'
      when cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))<=3000 and cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))>1000 then 'B'
      when cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))<=1000 and cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))>200 then 'C'
      when cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))<=200  and cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))>50  then 'D'
      when cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))<=50   and cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))>30 then 'E'
      when cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))<=30   and cast(t3.mm_waybill_cnt/t3.send_days  as decimal(16,0))>0 then 'F'
 end as  customer_level --客户等级
,t3.is_loss3        --是否连续三日流失
,t3.is_loss         --是否当日流失
,t3.is_new_send     --是否新增且开户客户
,t3.is_old_send     --是否新增已开户客户
,t3.send_days       --当月发货天数
,t3.mm_waybill_cnt  --当月发货量
--,is_new_mm_cnt   --当月新增客户数
--,is_send_mm_cnt  --当月发货客户数
,t3.is_send           --当日是否发货
,t3.ordersource_name  --订单来源
,net.agent_code    --代理区编码
,net.agent_name    --代理区名称
,net.fran_code     --加盟商编码
,net.fran_name     --加盟商名称
,t3.date_time   as dt   --查询日期
from
(
select
      t1.date_time
     ,t1.customer_code
     ,t1.customer_name
     ,t1.customer_create_time
     ,t1.staff_code
     ,t1.staff_name
     ,t1.belong_type
     ,t1.network_code
     ,t1.network_name
     ,t1.waybill_cnt                         --当日发货量
     ,t1.send_days                           --当月发货天数
     ,t1.mm_waybill_cnt                      --当月发货量
     ,if(t1.day3_send_cnt=0,1,0) as is_loss3 --是否连续三日流失
     ,if(t1.waybill_cnt=0,1,0) as is_loss    --是否当日流失
     ,if(t1.waybill_cnt>0 and date(t1.customer_create_time)='{{ execution_date | cst_ds}}',1,0) as is_new_send --是否新增且开户客户
     ,if(t1.three_send_cnt=0 and t1.waybill_cnt>0 and date(t1.customer_create_time) < date_add('{{ execution_date | cst_ds}}',-1),1,0 ) as is_old_send --是否新增已开户客户
    -- ,t2.is_new_mm_cnt
    -- ,t2.is_send_mm_cnt
     ,if(t1.waybill_cnt>0 and t1.date_time='{{ execution_date | cst_ds}}',1,0) as is_send --当日是否发货
     ,t1.ordersource_name
from
    (
        select
            date_time
             ,customer_code
             ,customer_name
             ,customer_create_time
             ,staff_code
             ,staff_name
             ,belong_type
             ,network_code
             ,network_name
             ,waybill_cnt  --当日发货量
             ,count(if(waybill_cnt>0,1,null)) over(partition by customer_code,staff_code,network_code,date_format(date_time,'yyyy-MM') order by date_time) as send_days    --当月发货天数
             ,sum(waybill_cnt) over(partition by customer_code,staff_code,network_code, date_format(date_time,'yyyy-MM') order by date_time) as mm_waybill_cnt  --当月发货量
             ,sum(waybill_cnt) over(PARTITION by customer_code,staff_code,network_code order by date_time rows BETWEEN 2 PRECEDING and CURRENT ROW) as day3_send_cnt   --连续三日货量
             ,sum(if(date_time between add_months(trunc('{{ execution_date | cst_ds}}','MM'),-3) and date_add('{{ execution_date | cst_ds}}',-1),waybill_cnt,0 )) over(partition by customer_code,staff_code,network_code ) as three_send_cnt --最近3个月发货量
             ,ordersource_name --订单来源
        from cust_way
    )t1
)t3 left join jms_dim.dim_network_whole_massage net on t3.network_code=net.code
where t3.date_time='{{ execution_date | cst_ds}}'
distribute by dt, abs(hash(customer_code)) % 3 ;
---------汇总--------------------------------------------
with  tab_cnt as (
select
    grouping__id
        ,date_time                               --日期
        ,staff_code                              --业务员code
        ,max(staff_name) as staff_name                             --业务员
        ,max(belong_type) as belong_type                              --所属类别
        ,network_code                            --网点编码
        ,max(network_name) as network_name                          --网点名称
        ,customer_level                          --客户等级
        ,ordersource_name
--,if(grouping__id in (1,2),count(1)            ,0) as customer_cnt     --客户数
        ,case when grouping__id=1 then count(1)
    when grouping__id=2 then count(if(waybill_cnt>0,1,null))
    end as customer_cnt
        ,if(grouping__id=3,max(is_send_mm_cnt) ,0) as mm_send_cust_cnt --月发货客户数
        ,if(grouping__id=3,sum(is_send)        ,0) as send_cust_cnt    --当日发货客户数
        ,if(grouping__id=3,sum(is_new_send)    ,0) as new_cust_cnt     --当日新增且开户客户数
        ,if(grouping__id=3,sum(is_old_send)    ,0) as old_cust_cnt     --当日新增且已开户客户数
        ,if(grouping__id=3,max(is_new_mm_cnt)  ,0) as mm_newcust_cnt   --当月新增客户数
        ,if(grouping__id=3,sum(is_loss)        ,0) as loss_cust_cnt    --当日流失客户数
        ,if(grouping__id=3,sum(is_loss3)       ,0) as loss3_cust_cnt   --连续3日流失客户数  
        ,case when grouping__id=1 then 1  --等级
    when grouping__id=3 then 2  --业务员维度
    when grouping__id=2 then 3  --订单来源维度
    end as dim_type
from
    (
         select
         d1.date_time
        ,d1.customer_code
        ,d1.customer_name
        ,d1.customer_create_time
        ,d1.staff_code
        ,d1.staff_name
        ,d1.belong_type     --所属类别
        ,d1.network_code
        ,d1.network_name
        ,d1.waybill_cnt     --当日发货量
        ,d1.avg_waybill_cnt  --日均发货
        ,d1.customer_level
        ,d1.is_loss3        --是否连续三日流失
        ,d1.is_loss         --是否当日流失
        ,d1.is_new_send     --是否新增且开户客户
        ,d1.is_old_send     --是否新增已开户客户
        ,d1.send_days       --当月发货天数
        ,d1.mm_waybill_cnt  --当月发货量
        ,d2.is_new_mm_cnt   --当月新增客户数
        ,d2.is_send_mm_cnt  --当月发货客户数
        ,d1.is_send         --当日是否发货
        ,d1.ordersource_name --订单来源
        from jms_dm.dm_network_customer_analysis_detail_dt d1
        left join
        (--网点和业务员为空对应客户数据量有点大，开窗去重会发生数据倾斜，所以单独处理
            select
                  date_format(date_time,'yyyy-MM') as mm_time
                 ,staff_code
                 ,network_code
                 ,count(distinct if(is_new_send=1 or is_old_send=1,customer_code,null)) as is_new_mm_cnt --当月新增客户数
                 ,count(distinct if(is_send=1 ,customer_code,null))     as is_send_mm_cnt --当月发货客户数
            from jms_dm.dm_network_customer_analysis_detail_dt d1
            where dt between trunc('{{ execution_date | cst_ds}}','MM') and '{{ execution_date | cst_ds}}'
            group by staff_code,network_code,date_format(date_time,'yyyy-MM')
        )d2 on d1.staff_code=d2.staff_code and d1.network_code=d2.network_code and date_format(d1.date_time,'yyyy-MM')=d2.mm_time and d1.dt='{{ execution_date | cst_ds}}'
        where d1.dt = '{{ execution_date | cst_ds}}'
  )t4 group by date_time,staff_code,network_code,customer_level,ordersource_name
grouping sets((date_time,staff_code,network_code),(date_time,staff_code,network_code,customer_level),(date_time,network_code,staff_code,ordersource_name))
--分别对应 grouping__id （3 业务员，1 等级，2订单来源）
)

insert overwrite table jms_dm.dm_network_customer_analysis_summary_dt partition(dt)
select
    a.date_time                               --日期
     ,a.network_code                            --网点编码
     ,a.network_name                            --网点名称
     ,a.staff_code                              --业务员code
     ,a.staff_name                              --业务员
     ,a.belong_type                             --所属类别
     ,a.customer_level   --客户等级
     ,case when ordersource_name ='326'  then '桃花岛'
           when ordersource_name ='348'  then '紫金山'
           when ordersource_name ='352'  then '逍遥峰'
           when ordersource_name ='353'  then '七星潭'
           when ordersource_name ='328'  then '极地湾'
           when ordersource_name ='其他' then '其他'
    end as ordersource_name
     ,a.customer_cnt     --客户数
     ,a.mm_send_cust_cnt --月发货客户数
     ,a.send_cust_cnt    --当日发货客户数
     ,a.new_cust_cnt     --当日新增且开户客户数
     ,a.old_cust_cnt     --当日新增且已开户客户数
     ,a.mm_newcust_cnt   --当月新增客户数
     ,a.loss_cust_cnt    --当日流失客户数
     ,a.loss3_cust_cnt   --连续3日流失客户数 
     ,a.dim_type
     ,net.agent_code
     ,net.agent_name
     ,net.fran_code
     ,net.fran_name
     ,a.date_time as dt
from tab_cnt a
left join jms_dim.dim_network_whole_massage net on a.network_code=net.code
where a.dim_type in (2,3) or (a.dim_type=1 and a.customer_level is not null)
distribute by 1;